package com.dd.keel.core.common.flea;

import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.io.IOException;
import java.io.OutputStream;
import java.io.InputStream;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.dd.keel.core.common.utils.*;

import java.util.Date;
import java.util.concurrent.atomic.AtomicBoolean;


public class TCPEndpoint{
	
	final static private Logger LOGGER = LoggerFactory.getLogger(TCPEndpoint.class);
	
	private AtomicBoolean currentStatus = new AtomicBoolean();
	
	private String serverIP = null;
	private int serverPort = 0;
	private Socket client = null;
	private OutputStream out = null;
	private InputStream in = null;
	
	private boolean isClient = false;
	private int max_connect_retry = 2;
	
	private String protocolKey = ProtocolCodecFactory.KEY_DEFAULT;
	
	private boolean keepAlive = true;
	
	private String localAddress = null;
	private int localPort = -1;
	private int connectTimeoutSeconds = 5;
	
	private String connectionName = null;
	
	public TCPEndpoint(String serverIP,int serverPort,String protocolKey)
	{
		this.serverIP = serverIP;
		this.serverPort = serverPort;
		if(protocolKey!=null) this.protocolKey = protocolKey;
		isClient = true;
	}
	
	public TCPEndpoint(String serverIP,int serverPort)
	{
		this.serverIP = serverIP;
		this.serverPort = serverPort;
		isClient = true;
	}
	
	public TCPEndpoint(String localAddress,int localPort,String remoteAddress,int remotePort)
	{
		this.localAddress = localAddress;
		this.localPort = localPort;
		this.serverIP = remoteAddress;
		this.serverPort = remotePort;
		isClient = true;
	}
	
	public TCPEndpoint(Socket client)
	{
		this.client = client;
	}
	
	public void setStatus(boolean status)
	{
		currentStatus.set(status);
	}
	
	public boolean getStatus()
	{
		return currentStatus.get();
	}
	
	public String toString()
	{
		return this.connectionName;
	}
	
	public boolean connect() throws ConnectException
	{
		if(client==null)
		{
			boolean flag = true;
			try{
				LOGGER.debug( "begin connect to {}:{} ", serverIP, serverPort );
				client = new Socket();
				client.setKeepAlive(keepAlive);
				client.setReceiveBufferSize(37*1024);
				client.setSendBufferSize(25*1024);
				client.setTcpNoDelay(true);
				client.setReuseAddress(true);
				client.setSoLinger(true, 30);
				SocketAddress localAddr = null;
				if(localAddress!=null && localPort!=-1)
				{
					localAddr = new InetSocketAddress(localAddress,localPort); 
				}
				if(localAddr!=null)
				{
					client.bind(localAddr);
				}
				SocketAddress remoteAddr = new InetSocketAddress(serverIP,serverPort); 
				client.connect(remoteAddr,connectTimeoutSeconds*1000);
				connectionName = "LOCAL="+client.getLocalAddress()+":"+client.getLocalPort()+"<->REMOTE="+client.getInetAddress()+":"+client.getPort();
				setStatus(true);
				out = client.getOutputStream();
				in = client.getInputStream();
				//client.setKeepAlive(true);
				flag = isConnected();
				if(flag)
					LOGGER.debug( "connect to {}:{} successfully!", serverIP, serverPort );
				else
					LOGGER.warn( "connect to {}:{} failed!", serverIP, serverPort );
			}catch(IOException ex){
				LOGGER.error( ex.getMessage(), ex );
				client = null;
				connectionName = null;
				setStatus(false);
				flag = false;
				throw new ConnectException(ex.getMessage());
			}catch(Exception ex){
				LOGGER.error( ex.getMessage(), ex );
				client = null;
				connectionName = null;
				setStatus(false);
				flag = false;
				throw new ConnectException(ex.getMessage());
			}
			return flag;
		}
		else
		{
			return true;
		}
	}
	
	public void disconnect()
	{
		LOGGER.debug( "close to {}:{}", serverIP, serverPort );
		try{
			if(out!=null) out.close();
		}catch(Exception ex){
			LOGGER.error( ex.getMessage(), ex );
		}
		try{
			if(in!=null) in.close();
		}catch(Exception ex){
			LOGGER.error( ex.getMessage(), ex );
		}
		try{
			if(client!=null) client.close();
		}catch(Exception ex){
			LOGGER.error( ex.getMessage(), ex );
		}
		client = null;
		connectionName = null;
	}
	
	public boolean isConnected()
	{
		if(client!=null)
		{
			return client.isConnected() && !client.isClosed();
		}else{
			return false;
		}
	}
	
	public boolean reconnect() throws ConnectException
	{
		boolean connected = false;
		if(isClient)
		{
			disconnect();
			int retry = 0;
			while( !connected && retry<max_connect_retry )
			{
				LOGGER.debug( "begin reconnect to {}:{}", serverIP, serverPort );
				try{
					client = new Socket();
					client.setKeepAlive(keepAlive);
					client.setReceiveBufferSize(37*1024);
					client.setSendBufferSize(25*1024);
					client.setTcpNoDelay(true);
					client.setReuseAddress(true);
					client.setSoLinger(true, 30);
					SocketAddress localAddr = null;
					if(localAddress!=null && localPort!=-1)
					{
						localAddr = new InetSocketAddress(localAddress,localPort); 
					}
					if(localAddr!=null)
					{
						client.bind(localAddr);
					}
					SocketAddress remoteAddr = new InetSocketAddress(serverIP,serverPort); 
					client.connect(remoteAddr,connectTimeoutSeconds*1000);
					connectionName = "LOCAL="+client.getLocalAddress()+":"+client.getLocalPort()+"<->REMOTE="+client.getInetAddress()+":"+client.getPort();
					setStatus(true);
					out = client.getOutputStream();
					in = client.getInputStream();
					//
					connected = true;
					LOGGER.debug( "reconnect to {}:{} successfully!", serverIP, serverPort );
				}catch(Exception ex){
					LOGGER.warn( "reconnect to {}:{} failed!", serverIP, serverPort );
					LOGGER.error( ex.getMessage(), ex );
					setStatus(false);
					connectionName = null;
					throw new ConnectException(ex.getMessage());
				}
				try{
					Thread.sleep(DateUtil.randomTime());
				}catch(Exception ex){
					LOGGER.error( ex.getMessage(), ex );
				}
				++retry;
			}
		}
		return connected;
	}
	
	public void send(byte[] buffer) throws ConnectException,SendException  
	{
		send(buffer,0,buffer.length);
	}
	
	public void send(byte[] buffer,int off,int len) throws ConnectException,SendException 
	{
		if(client==null) connect();
		try{
			if(out==null) out = client.getOutputStream();
		}catch(IOException ex){
			throw new SendException(ex.getMessage());
		}
		try{
			if(LOGGER.isTraceEnabled())
			{
				LOGGER.trace("Sending::: {}",ByteUtil.byte2hexWithBlank(buffer));
			}
			out.write(buffer, off, len);
			out.flush();
		}catch(IOException ex){
			throw new SendException(ex.getMessage());
		}
		//lastSendDatetime = new Date();
	}
	
	public synchronized void send(FleaMessage message) throws ConnectException,EncodeException,SendException 
	{
		ProtocolEncoder encoder = null;
		try {
			encoder = ProtocolCodecFactory.getEncoder(protocolKey);
		} catch (Exception e) {
			throw new SendException(e.getMessage());
		}
		LOGGER.trace( "send message to *** {}", toString() );
		encoder.clean();
		send(encoder.encode(message));
	}
	
	public synchronized void receive(FleaMessage message) throws ConnectException,ReceiveException,DecodeException
	{
		ProtocolDecoder decoder = null;
		try {
			decoder = ProtocolCodecFactory.getDecoder(protocolKey);
		} catch (Exception e) {
			throw new ReceiveException(e.getMessage());
		}
		LOGGER.trace( "receive message from *** {}", toString() );
		if(client==null) connect();
		try{
			if(in==null) in = client.getInputStream();
		}catch(IOException ex){
			throw new ReceiveException(ex.getMessage());
		}
		try {
			client.setSoTimeout(0);
		} catch (SocketException e) {
			throw new ConnectException(e.getMessage());
		}
		decoder.clean();
		try{
			decoder.decode(in,message);
			LOGGER.trace( "have received message from *** {}", toString() );
		}catch(DecodeException ex){
			LOGGER.error( "not received message from *** {} ; Error : {}", toString(), ex.getMessage() );
			throw ex;
		}catch(ConnectException ex){
			LOGGER.error( "not received message from *** {} ; Error : {}", toString(), ex.getMessage() );
			throw ex;
		}
	}
	
	public synchronized void receive(FleaMessage message,int timeoutSeconds) throws ConnectException,ReceiveException,DecodeException
	{
		ProtocolDecoder decoder = null;
		try {
			decoder = ProtocolCodecFactory.getDecoder(protocolKey);
		} catch (Exception e) {
			throw new ReceiveException(e.getMessage());
		}
		LOGGER.trace( "receive message from *** {}", toString() );
		if(client==null) connect();
		try{
			if(in==null) in = client.getInputStream();
		}catch(IOException ex){
			throw new ReceiveException(ex.getMessage());
		}
		try {
			client.setSoTimeout(timeoutSeconds*1000);
		} catch (SocketException e) {
			throw new ConnectException(e.getMessage());
		}
		decoder.clean();
		try{
			decoder.decode(in,message);
			LOGGER.trace( "have received message from *** {}", toString() );
		}catch(DecodeException ex){
			LOGGER.error( "not received message from *** {} ; Error : {}", toString(), ex.getMessage() );
			throw ex;
		}catch(ConnectException ex){
			LOGGER.error( "not received message from *** {} ; Error : {}", toString(), ex.getMessage() );
			throw ex;
		}
	}
	
	public InputStream getInputStream() throws ConnectException,IOException
	{
		if(client==null) connect();
		try{
			if(in==null) in = client.getInputStream();
			return in;
		}catch(IOException ex){
			throw ex;
		}
	}
	
	public OutputStream getOutputStream() throws ConnectException,IOException
	{
		if(client==null) connect();
		try{
			if(out==null) out = client.getOutputStream();
			return out;
		}catch(IOException ex){
			throw ex;
		}
	}
	
	public int available() throws java.io.IOException
	{
		return in.available();
	}
	
	/**
	 * The entity needs to be consumed completely in order to re-use the connection for the next request/response
	 * @throws java.io.IOException
	 */
	public void consumeContent() throws java.io.IOException
	{
		if(in!=null)
		{
			if(in.available()>0) in.skip(in.available());
		}
	}

	public boolean isKeepAlive() {
		return keepAlive;
	}

	public void setKeepAlive(boolean keepAlive) {
		this.keepAlive = keepAlive;
	}

	public long getSoTimeoutSeconds() throws SocketException {
		return client.getSoTimeout()/1000;
	}
	
	public void setSoTimeoutSeconds(long soTimeoutSeconds) throws SocketException {
		if(client!=null) client.setSoTimeout((int)soTimeoutSeconds*1000);
	}

	public int getConnectTimeoutSeconds() {
		return connectTimeoutSeconds;
	}

	public void setConnectTimeoutSeconds(int connectTimeoutSeconds) {
		this.connectTimeoutSeconds = connectTimeoutSeconds;
	}
	
}
